package org.my.hbase.test;
import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;


public class WordCountHBase {
    public static class Map extends Mapper<Object, Text, Text, IntWritable>{
    	private IntWritable i = new IntWritable(1);
    	public void map(Object key,Text value,Context context) throws IOException, InterruptedException{
    		StringTokenizer line = new StringTokenizer(value.toString());
    		while(line.hasMoreTokens()){
    			String val = line.nextToken();
    			context.write(new Text(val), i);
    		}
    	}
    }
    
    public static class Reduce extends TableReducer <Text, IntWritable, ImmutableBytesWritable>{
    	public void reduce(Text key,Iterable<IntWritable> values,Context context) throws IOException, InterruptedException{
    		int sum = 0;
    		for(IntWritable val:values){
    			sum+=val.get();
    		}
    		Put put = new Put(Bytes.toBytes(key.toString()));
    		put.add(Bytes.toBytes("content"),Bytes.toBytes("count"),Bytes.toBytes(String.valueOf(sum)));
    		context.write(new ImmutableBytesWritable(key.getBytes()), put);
    	}
    }
    
    public static void createHBaseTable(String tableName) throws Exception{
    	HTableDescriptor thd = new HTableDescriptor(tableName);
    	HColumnDescriptor col = new HColumnDescriptor("content");
    	thd.addFamily(col);
    	HBaseConfiguration configuration = new HBaseConfiguration();
    	HBaseAdmin admin = new HBaseAdmin(configuration);
    	if(admin.tableExists(tableName)){
    		System.out.println("tableName exits,trying rescreate table");
    		admin.disableTable(tableName);
    		admin.deleteTable(tableName);
    	}
    	System.out.println("create new table:"+tableName);
    	admin.createTable(thd);
    }
    
    public static void main(String[] args) throws Exception {
		String tableName = "wordcount";
		Configuration conf = new Configuration();
		conf.set(TableOutputFormat.OUTPUT_TABLE, tableName);
		createHBaseTable(tableName);
		String input = args[0];
		Job job = new Job(conf,"WordCount table "+input);
		job.setJarByClass(WordCountHBase.class);
		job.setMapperClass(Map.class);
		TableMapReduceUtil.initTableReducerJob(tableName, Reduce.class, job);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		FileInputFormat.addInputPath(job, new Path(args[0]));
		System.exit(job.waitForCompletion(true)?0:1);
	}
}
